package org.aurora.client;

import org.apache.commons.lang3.StringUtils;
import org.aurora.client.exception.AuroraTaskConnectSchedulerServerException;
import org.aurora.client.processor.ExecuteTaskProcessor;
import org.aurora.common.util.HttpClientUtil;
import org.aurora.common.util.ThreadUtil;
import org.aurora.model.request.ReportTaskRequest;
import org.aurora.model.response.ClientFetchSchedulerServerAddrResponse;
import org.aurora.model.response.CommonResponse;
import org.aurora.remote.netty.NettyRemoteClient;
import org.aurora.model.type.RequestCommandType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.*;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

public class DefaultAuroraClient implements AuroraClient {
    private static final Logger log = LoggerFactory.getLogger(DefaultAuroraClient.class);

    private NettyRemoteClient remoteClient;

    private AuroraClientConfig config;

    private Map<String, ITaskProcessor> taskProcessorMap;

    private ExecuteTaskProcessor executeTaskProcessor;

    private final AtomicBoolean isStart = new AtomicBoolean(false);
    private final AtomicBoolean isStop = new AtomicBoolean(false);

    public DefaultAuroraClient(AuroraClientConfig config, Map<String, ITaskProcessor> taskProcessorMap) {
        validateConfig(config);
        this.config = config;
        this.taskProcessorMap = taskProcessorMap;
        this.executeTaskProcessor = new ExecuteTaskProcessor(this.config.getCoreThreadSize(), this.config.getMaxThreadSize(), new HashMap<>(this.taskProcessorMap));
    }

    private void validateConfig(AuroraClientConfig config) {
        if (config == null) {
            throw new IllegalArgumentException("Aurora client config cannot be null");
        }

        String fetchSchedulerServerAddrUrl = config.getFetchSchedulerServerAddrUrl();
        int maxThreadSize = config.getMaxThreadSize();
        int coreThreadSize = config.getCoreThreadSize();

        if (coreThreadSize <= 0) {
            throw new IllegalArgumentException("Aurora client config coreThreadSize must be greater than 0");
        }
        if (maxThreadSize <= 0) {
            throw new IllegalArgumentException("Aurora client config maxThreadSize must be greater than 0");
        }
        if (maxThreadSize < coreThreadSize) {
            throw new IllegalArgumentException("Aurora client config maxThreadSize must be greater than or equal to coreThreadSize");
        }
        if (StringUtils.isBlank(fetchSchedulerServerAddrUrl)) {
            throw new IllegalArgumentException("Aurora client config fetchSchedulerServerAddrUrl cannot be blank");
        }
    }

    @Override
    public void start() {
        if (isStart.compareAndSet(false, true)) {
            // 获取当前任务
            // 连接到 scheduler 服务端并上报任务,启动后不可重新添加任务
            this.connectSchedulerServer();
        }
    }

    public synchronized void connectSchedulerServer() {
        String fetchSchedulerServerAddrUrl = this.config.getFetchSchedulerServerAddrUrl();
        long fetchSchedulerServerAddrTimeout = this.config.getFetchSchedulerServerAddrTimeout();
        int fetchSchedulerServerAddrRetryTimes = this.config.getFetchSchedulerServerAddrRetryTimes();
        long fetchSchedulerServerAddrRetryInterval = this.config.getFetchSchedulerServerAddrRetryInterval();
        String appcode = this.config.getAppcode();

        int currentFetchSchedulerServerAddrTimes = 0;
        do {
            try {
                ClientFetchSchedulerServerAddrResponse fetchSchedulerServerAddrResponse = HttpClientUtil.get(
                        fetchSchedulerServerAddrUrl,
                        Collections.singletonMap("appcode", appcode),
                        ClientFetchSchedulerServerAddrResponse.class,
                        fetchSchedulerServerAddrTimeout, TimeUnit.MILLISECONDS);

                currentFetchSchedulerServerAddrTimes++;

                if (fetchSchedulerServerAddrResponse != null) {
                    String schedulerServerIp = fetchSchedulerServerAddrResponse.getSchedulerServerIp();
                    int schedulerServerPort = fetchSchedulerServerAddrResponse.getSchedulerServerPort();

                    this.remoteClient = new NettyRemoteClient(schedulerServerIp, schedulerServerPort);
                    this.remoteClient.registerProcessor(RequestCommandType.EXECUTE_TASK.getCode(), this.executeTaskProcessor);

                    this.remoteClient.start();

                    reportTask();
                    return;
                }
                log.error("Failed to fetch scheduler server address, response is null");
            } catch (Exception e) {
                if (this.remoteClient != null) {
                    this.remoteClient.shutdown();
                }
                log.error("Failed to connect scheduler server address", e);
            }
            // 递增睡眠
            ThreadUtil.sleepIgnoreInterrupt(currentFetchSchedulerServerAddrTimes * fetchSchedulerServerAddrRetryInterval);
        } while (currentFetchSchedulerServerAddrTimes < fetchSchedulerServerAddrRetryTimes);

        // 连接失败抛出异常
        throw new AuroraTaskConnectSchedulerServerException();
    }

    private void reportTask() {
        long reportTaskTimeout = this.config.getReportTaskTimeout();

        String appcode = this.config.getAppcode();
        List<String> taskNames = new ArrayList<>(this.taskProcessorMap.keySet());

        ReportTaskRequest request = new ReportTaskRequest(appcode, taskNames);

        CommonResponse response = this.remoteClient.send(request, reportTaskTimeout, CommonResponse.class);

        if (response == null) {
            throw new AuroraTaskConnectSchedulerServerException("Failed to send report task, response is null");
        }

        // todo 指定异常直接中断
        if (!response.isSuccess()) {
            throw new AuroraTaskConnectSchedulerServerException("Failed to send report task, response fail, error msg: " + response.getErrorMsg());
        }
    }

    @Override
    public void stop() {
        // 添加钩子函数结束时断开连接清除状态
        if (isStart.get() && isStop.compareAndSet(false, true)) {
            if (this.remoteClient != null) {
                this.remoteClient.shutdown();
            }
        }
    }
}
